package cn._51doit

import java.util.{Properties, UUID}

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

/**
 * ┏┓　　　┏┓
 * ┏┛┻━━━┛┻┓
 * ┃　　　　　　　┃ 　
 * ┃　　　━　　　┃
 * ┃　┳┛　┗┳　┃
 * ┃　　　　　　　┃
 * ┃　　　┻　　　┃
 * ┃　　　　　　　┃
 * ┗━┓　　　┏━┛
 *    　　┃　　　┃神兽保佑
 *    　　┃　　　┃代码无BUG！
 *    　　┃　　　┗━━━┓
 *    　　┃　　　　　　　┣┓
 *    　　┃　　　　　　　┏┛
 *    　　┗┓┓┏━┳┓┏┛
 *    　　　┃┫┫　┃┫┫
 *    　　　┗┻┛　┗┻┛
 *
 * @create: 2021-10-11 22:38
 * @author: 今晚打脑斧
 * @program: kafka
 * @Description:
 * 用来测试生产者写入kafka的方式
 *   生成者将数据写入Kafka只有轮询的策略吗？
 *    1、可以将数据写入到指定的分区中，必须指定分区编号
 *    2、如果指定了分区编号，同时又指定了非null的key，优先使用分区编号
 *    3、如果没有指定分区编号，指定了非null的key，可以保证相同key的是一定写入到同一个分区，但是同一分区中可能有多个不同的key（类似HashPartitioner）
 **/
object ScalaProducer3 {
  def main(args: Array[String]): Unit = {
    //指定broker的地址
    val properties = new Properties
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "doit01:9092,doit02:9092,doit03:9092")
    //指定key序列化方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    properties.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //创建一个生产者
    val producer = new KafkaProducer[String, String](properties)

    // 1
//        val producerRecord1 = new ProducerRecord[String, String]("wordcount", 0, null ,"hello kafka 1")
//        producer.send(producerRecord1)
//
//        val producerRecord2 = new ProducerRecord[String, String]("wordcount", 0, null, "hello kafka 2")
//        producer.send(producerRecord2)
//
//        val producerRecord3 = new ProducerRecord[String, String]("wordcount", 1, null, "hello kafka 3")
//        producer.send(producerRecord3)
//
//        producer.flush()
//
//        val producerRecord4 = new ProducerRecord[String, String]("wordcount", 2, null, "hello kafka 3")
//        producer.send(producerRecord4)

      // 2
//        val producerRecord1 = new ProducerRecord[String, String]("wordcount", 1, "spark", "hello kafka 1")
//        producer.send(producerRecord1)
//
//        val producerRecord2 = new ProducerRecord[String, String]("wordcount",1, "hadoop", "hello kafka 2")
//        producer.send(producerRecord2)
//
//        val producerRecord3 = new ProducerRecord[String, String]("wordcount", 0,"spark", "hello kafka 3")
//        producer.send(producerRecord3)
//
//        val producerRecord4 = new ProducerRecord[String, String]("wordcount", 1,"flink", "hello kafka 4")
//        producer.send(producerRecord4)
//
//        val producerRecord5 = new ProducerRecord[String, String]("wordcount", 2, "flink", "hello kafka 5")
//        producer.send(producerRecord5)

    // 3
    for(i <- 1 to 20) {
      //随机生成UUID
      val uuid = UUID.randomUUID().toString
      val producerRecord5 = new ProducerRecord[String, String]("wordcount", uuid, "hello kafka 5")
      producer.send(producerRecord5)

    }

    producer.close()
  }
}
